package com.shujia.tf

import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import com.shujia.bean._
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

class YongDuWindow extends ProcessWindowFunction[CarWide, YDBean, String, TimeWindow] {
  override def process(key: String, context: Context, elements: Iterable[CarWide], out: Collector[YDBean]): Unit = {

    val end = context.window.getEnd


    val dl = key.split("-")(0)
    val kc = key.split("-")(1)


    //计算车流量和平均速度
    var flow = 0L
    var sum = 0.0
    for (elem <- elements) {
      flow += 1
      sum += elem.getSpeed
    }

    val avgSpeed = sum / flow


    /**
      *
      *
      * 1、绿色
      * 车辆数量小于100
      * 平均速度45以上
      *
      * 2、黄色
      * 车辆数量大于100
      * 平均速度40-45
      *
      * 3、红色
      * 车辆数量大于100
      * 平均速度35-40
      *
      * 4、深红
      * 车辆数量大于100
      * 平均速度小于35
      *
      */

    val yd = new YDBean
    yd.setDl(dl)
    yd.setKc(kc)
    yd.setTs(end)

    if (flow <= 100 | avgSpeed > 45) {
      yd.setState("绿色")
      out.collect(yd)
    } else if (flow > 100 & avgSpeed > 40 & avgSpeed <= 45) {
      yd.setState("黄色")
      out.collect(yd)
    } else if (flow > 100 & avgSpeed > 35 & avgSpeed <= 40) {
      yd.setState("红色")
      out.collect(yd)
    } else if (flow > 100 & avgSpeed <= 35) {
      yd.setState("深红色")
      out.collect(yd)
    }
  }
}
